use pulsar::{Pulsar, TokioExecutor, producer, SerializeMessage, Producer, Error as PulsarError, Authentication};
use crate::common::config::PulsarConfig;
use std::sync::Arc;

pub struct PulsarBus {
    pulsar: Pulsar<TokioExecutor>,
}

impl PulsarBus {
    pub async fn new(cfg: &PulsarConfig) -> Result<Self, PulsarError> {
        let url = cfg.url.clone();
        let mut builder = Pulsar::builder(url, TokioExecutor);
        // 可选鉴权（token）
        if let Some(token) = get_pulsar_token(cfg) {
            builder = builder.with_auth(Authentication::token(token));
        }
        let pulsar: Pulsar<_> = builder.build().await?;
        Ok(Self { pulsar })
    }

    pub async fn create_producer(&self, topic: &str) -> Result<Producer<TokioExecutor>, PulsarError> {
        self.pulsar.create_producer(topic, None).await
    }
}

fn get_pulsar_token(cfg: &PulsarConfig) -> Option<String> {
    // 优先环境变量 PULSAR_TOKEN，其次配置文件字段（若存在）
    if let Ok(token) = std::env::var("PULSAR_TOKEN") { return Some(token); }
    #[allow(dead_code)]
    struct TokenAccessor;
    // 由于当前 PulsarConfig 未定义 token 字段，这里仅从环境变量读取
    // 如需从配置读取，可在 PulsarConfig 增加: pub token: Option<String>
    None
}

impl SerializeMessage for Vec<u8> {
    fn serialize_message(input: &Self) -> Result<pulsar::producer::Message, pulsar::Error> {
        Ok(pulsar::producer::Message { payload: input.clone(), ..Default::default() })
    }
}


